package com.zidongxiangxi.clock.server.job;

import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.google.common.collect.Lists;
import com.zidongxiangxi.clock.server.business.ClockTaskBusiness;
import com.zidongxiangxi.clock.server.config.ClockConfig;
import com.zidongxiangxi.clock.server.constants.RedisConstant;
import com.zidongxiangxi.clock.server.repository.ClockDao;
import com.zidongxiangxi.clock.server.repository.entity.Clock;
import com.zidongxiangxi.clock.server.utils.ClockUtils;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * 加载闹钟工作线程
 *
 * @author chenxudong
 * @date 2020/07/22
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class LoadClockWorkThread extends AbstractWorkThread {
    private final ClockConfig config;

    private final ClockDao clockManger;
    private final ClockTaskBusiness clockTaskBusiness;
    private final RedissonClient redissonClient;

    /**
     * 工作线程是否停止
     */
    private volatile boolean stop = false;

    @Override
    public void run() {
        log.info("start {} ........", this.getClass().toString());
        int nextMaxTaskNum = config.getDatabase().getLoadTaskMinBatchSize();
        while (!isStop()) {
            try {
                RLock lock = redissonClient.getLock(RedisConstant.LOAD_CLOCK_KEY);
                nextMaxTaskNum = loadClock(nextMaxTaskNum, lock);
            } catch (Exception e) {
                log.error("Load clock error, message =", e);
            }
        }
        log.info("stop {} ........", this.getClass());
    }

    private int loadClock(int nextMaxTaskNum, RLock lock) {
        long startMillis = System.currentTimeMillis();
        boolean taskNumMax = false;
        try {
            // 加分布式锁
            lock.lock();

            //从数据库加载任务
            long minNextTryTime = startMillis + config.getDatabase().getPreLoadMilliseconds();
            List<Clock> clocks = clockManger.loadAlarmClock(nextMaxTaskNum, minNextTryTime);
            if (CollectionUtils.isNotEmpty(clocks)) {
                //任务数达到最大值，说明还有其它任务会加载，不休眠
                if (clocks.size() == nextMaxTaskNum) {
                    taskNumMax = true;
                }
                //更新闹钟相关信息
                clocks = filterInvalidClocksAndUpdateNextTryInfo(clocks);
                //添加到任务队列
                clockTaskBusiness.addClockTask(clocks);
            }
        } catch (Exception e) {
            log.error("Load clock error, message =", e);
        } finally {
            // 释放分布式锁
            lock.unlock();
        }

        if (!taskNumMax) {
            // 没有达到最大任务数，表示任务已经加载完
            try {
                nextMaxTaskNum = config.getDatabase().getLoadTaskMinBatchSize();
                long sleepMillis = config.getDatabase().getLoadPeriodMilliseconds() - (System.currentTimeMillis() - startMillis);
                log.info("LoadAlarmClockWorkThread sleep {} millis ", sleepMillis);
                Thread.sleep(sleepMillis);
            } catch (InterruptedException e) {
                log.warn("LoadAlarmThreadWorkThread is interrupted !");
                Thread.currentThread().interrupt();
            }
        } else {
            // 达到最大任务数，不休眠，增大批次大小，继续加载数据
            nextMaxTaskNum = nextMaxTaskNum + config.getDatabase().getLoadTaskMinBatchSize();
            nextMaxTaskNum = Math.max(nextMaxTaskNum, config.getDatabase().getLoadTaskMaxBatchSize());
        }
        return nextMaxTaskNum;
    }

    @Override
    public void stop() {
        stop = true;
    }

    @Override
    public boolean isStop() {
        return stop;
    }

    /**
     * 更新闹钟相关信息：增加重试次数、删除多次发送闹钟
     *
     * @param clocks 闹钟
     */
    private List<Clock> filterInvalidClocksAndUpdateNextTryInfo(List<Clock> clocks) {
        if (CollectionUtils.isEmpty(clocks)) {
            return Lists.newLinkedList();
        }

        List<Clock> deleteClockIds = clocks.stream().filter(Objects::nonNull)
            .filter(clock -> clock.getRetryTimes() > config.getMaxRetryTimes())
            .collect(Collectors.toList());
        List<Clock> updateClocks = clocks.stream().filter(Objects::nonNull)
            .filter(clock -> clock.getRetryTimes() <= config.getMaxRetryTimes())
            .collect(Collectors.toList());
        if (CollectionUtils.isNotEmpty(updateClocks)) {
            List<Long> updateClockIds = updateClocks.stream().map(Clock::getId).collect(Collectors.toList());
            clockManger.changeNextTryTime(updateClockIds, ClockUtils.getNewNextTryTime());
        }
        if (CollectionUtils.isNotEmpty(deleteClockIds)) {
            clockManger.removeByIds(deleteClockIds);
        }
        return updateClocks;
    }

}
